package com.dagger.utils;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Future;

import javax.websocket.Session;

import com.alibaba.fastjson2.JSON;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class WebsocketUtil extends AbstractMsgUtil<Session> {

	private int DEFAULT_TIMEOUT = 500;

	/**
	 * 所有session集合
	 */
	private Map<String, Session> ONLINE_SESSION = new HashMap<>();

	/**
	 * 记录空闲Session集合
	 */
	private CopyOnWriteArraySet<Session> IDLE_SESSION = new CopyOnWriteArraySet<>();
	/**
	 * 记录正在使用中Session集合，value为Future，表示使用情况,正在发送消息
	 */
	private ConcurrentHashMap<Session, Future<Void>> BUSY_SESSION = new ConcurrentHashMap<>();

	/**
	 * 新增Session
	 *
	 * @param session session
	 */
	@Override
	public void _addUser(String msgKey, Session session) {
		addUser(session);
	}

	/**
	 * 关闭Session
	 *
	 * @param session session
	 */
	@Override
	public void _removeUser(Session session) {
		IDLE_SESSION.remove(session);
		BUSY_SESSION.remove(session);
	}

	@Override
	public boolean sendMessage(Session session, Object message) {
		return sendMessage(session, message, DEFAULT_TIMEOUT);
	}

	/**
	 * 发消息
	 *
	 * @param session session
	 * @param message message
	 * @param timeout timeout
	 */
	public synchronized boolean sendMessage(Session session, Object message, Integer timeout) {
		// timeout后放弃本次发送
		if (timeout < 0) {
			return true;
		}
		// 判断session是否空闲，抢占式
		if (IDLE_SESSION.remove(session)) {
			if (message instanceof String) {
				synchronized (session) {
					BUSY_SESSION.put(session, session.getAsyncRemote().sendText((String) message));
				}
			} else if (message instanceof ByteBuffer) {
				synchronized (session) {
					BUSY_SESSION.put(session, session.getAsyncRemote().sendBinary((ByteBuffer) message));
				}
			} else { // 消息类型有问题时，不发送。目前只添加了两种消息类型的发送方式，之后有需求可以扩展
				synchronized (session) {
					BUSY_SESSION.put(session, session.getAsyncRemote().sendText(JSON.toJSONString(message)));
				}
			}
			return true;
		} else {
			// 若session当前不在idle集合，则去busy集合中查看session上次是否已经发送完毕，即采用惰性判断
			synchronized (BUSY_SESSION) {
				if (BUSY_SESSION.containsKey(session) && BUSY_SESSION.get(session).isDone()) {
					BUSY_SESSION.remove(session);
					IDLE_SESSION.add(session);
				}
			}
			// 重试
			try {
				Thread.sleep(100);
				sendMessage(session, message, timeout - 100);
				return true;
			} catch (InterruptedException e) {
				log.error(e.getLocalizedMessage(), e);
				return false;

			}
		}
	}

	public void addUser(Session session) {
		IDLE_SESSION.add(session);
		ONLINE_SESSION.put(session.getId(), session);		
	}

//	public static synchronized void sendAll(Object message, Integer timeout) {
//		for (Session session : ONLINE_SESSION.values()) {
//			if (session.isOpen()) {
//				sendMessage(session, message, timeout);
//			}
//		}
//	}

//	public static synchronized void sendAll(Object message) {
//		// 重试5次
//		sendAll(message, DEFAULT_TIMEOUT);
//	}

	/**
	 * 获取对应类型下的会话中的参数
	 * 
	 * @param wsType
	 * @param key
	 * @return
	 */
//	public TreeSet<Object> getWsTypeProperties(String wsType, String key) {
//		TreeSet<Object> properties = new TreeSet<>();
//		for (Session session : ONLINE_SESSION.values()) {
//			if (session != null) {
//				Object wsTypeValue = session.getUserProperties().get("wsType");
//				Object value = session.getUserProperties().get(key);
//				if (wsType.equals(wsTypeValue)) {
//					properties.add(value);
//				}
//			}
//		}
//
//		return properties;
//	}

	/**
	 * 发送一个消息类型下的对应业务的消息
	 * 
	 * @param wsType
	 * @param key
	 * @param value
	 * @param message
	 */
//	public void sendMessage(String wsType, String key, Object value, String message) {
//		for (Session session : ONLINE_SESSION.values()) {
//			if (session != null) {
//				Object wsTypeValue = session.getUserProperties().get("wsType");
//				Object keyValue = session.getUserProperties().get(key);
//
//				if (wsType.equals(wsTypeValue) && value.equals(keyValue)) {
//					sendMessage(message, session);
//				}
//			}
//		}
//
//	}
}
